mike dfsu 处理

.dsfu 文件处理

• 读取指定数据[单个]

from mikeio import Dfsu

ps: 文件路径内不能有中文

filename = "xxx.dfsu"
item = 'Current speed'
step = -1

df = Dfus(filename)
all_data = df.read()

#获取 item 的step 数据，格式为 narray
data = all_data[item][step]

• 读取指定数据[多个]

# 实例化 dfsu 文件对象
df = Dfsu(filename)

# 根据指定类型获取
all_data = df.read(['U velocity','V velocity'])

# 获取指定时间序列
u_data = all_data[0][step][:]
v_data = all_data[1][step][:]

• 保存数据为xls、xlsx

import pandas as pd

xls_data = { "u":u_data, "v":v_data }

pd_data = pd.DataFrame(xls_data) # 转换数据为 DataFrame

writer = pd.ExcelWriter(filename) # 创建文件对象

pd_data.to_excel(writer, index=False) # 不添加索引

writer.save() 保存文件

• 获取mesh的网点

df = Dfsu(filename)
mesh = df.element_coordinates
X = mesh[:,0]
Y = mesh[:,1]
# Z = mesh[:,2] # 如果有

• 导出为shp

from mikeio import Dfsu,Dfs0,Mesh
import pandas as pd
import geopandas as gpd

filename = "xxx.dfsu"
item = 'Current speed'
step = -1

df = Dfsu(filename)
data = dfsu.read()

# 需要导出的数据
item_data = data[item][setp]
item_dataframe = pd.DataFrame({'speed': item_data})

# 获取shp格式
shp = dfsu.to_shapely()
poly_list = [e for e in shp] #???

# 转换为地理数据格式
gdf = gpd.GeoDataFrame(item_dataframe, geometry=poly_list)

# 导出shp
gdf.to_file("xxx.shp")

• 流速计算

• 流向计算

.xyz文件导出为.shp

import os
import mikeio as mio
import geopandas as gpd


def xyz_to_shp(input_file: str, output_file: str = None, overwrite: bool = True) -> str:
    """
    将.xyz文件导出为.shp文件

    参数:
    input_file: str - 输入的.xyz文件路径
    output_file: str - 输出的.shp文件路径（可选）
    overwrite: bool - 是否覆盖现有导出文件（默认为True）

    返回值:
    str - 成功导出时返回导出文件的绝对路径，失败时返回空字符串
    """
    try:
        # 读取.xyz文件
        data = mio.read_xyz(input_file)

        # 重命名列名称为小写的"x"、"y"和"z"
        data = data.rename(columns={"X": "x", "Y": "y", "Z": "z"})

        # 创建包含点数据的GeoDataFrame
        geometry = gpd.points_from_xy(data["x"], data["y"], data["z"])
        gdf = gpd.GeoDataFrame(data, geometry=geometry)

        # 导出为.shp文件
        if output_file is None:
            # 原地导出
            file_name, _ = os.path.splitext(input_file)
            output_file = file_name + ".shp"

        if os.path.exists(output_file):
            if not overwrite:
                print(f"输出文件 '{output_file}' 已存在，设置 overwrite 参数为 True 可以覆盖现有文件。")
                return os.path.abspath(output_file)
            # else:
            #     os.remove(output_file)

        gdf.to_file(output_file, driver="ESRI Shapefile")

        # 返回导出文件的绝对路径
        return os.path.abspath(output_file)
    except Exception as e:
        print("导出错误:", str(e))
        return ""


if __name__ == "__main__":
    shp_file = r"W:\CPS\MyProject\python-tools\py-tool\test\data\01_be_10.shp"
    xyz_files = [r"./流速流向采样点 .xyz", r"./水位采样点 .xyz"]

    for e in xyz_files:
        res = xyz_to_shp(e)
        print("res: ", res)